package com.tang.partition;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 分区Demo
 *
 * @author tang
 * @since 2023/6/8 16:48
 */
public class PartitionDemo {

    public static void main(String[] args) throws Exception {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(4);

        DataStreamSource<String> source = env.socketTextStream("192.168.70.141", 7777);
        
        // shuffle随机分区: random.nextInt(下游算子并行度)
//        source.shuffle().print();

        /*
         * rebalance轮询：nextChannelToSendTo = (nextChannelToSendTo + 1) % 下游算子并行度
         * 如果是数据倾斜场景，source后，调通rebalance，就可以解决数据倾斜问题
         */
//        source.rebalance().print();
        
        // rescale缩放，实现轮询局部组队，比rebalance更加高效
//        source.rescale().print();

        // 广播，发送给下游所有子任务
        source.broadcast().print();
        
        // 全部发送给第一个子任务。
//        source.global().print();

        // keyby: 按指定key去发送，相同key发往同一个子任务
        // one-to-one: Forward分区器
        
        // 总结：Flink提供了7中分区器和1中自定义分区器。
        
        env.execute();
    }
    
}
